
package models.spider.actor;

import java.util.concurrent.TimeUnit;

import models.aaa.File;

import play.libs.Akka;
import scala.concurrent.duration.FiniteDuration;
import akka.actor.Actor;
import akka.actor.ActorRef;
import akka.actor.Cancellable;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.routing.CurrentRoutees;

//import akka.util.FiniteDuration;

/**
 * Root actor for harvesting. Creates spiders(s)/needed actors and manage them.
 * 
 * @author laurent@opprecht.info
 */
public class Crawler extends UntypedActor {

    /**
     * Creates spiders actors.
     * 
     * @author laurent@opprecht.info
     * 
     */
    public static class SpiderFactory implements UntypedActorFactory {

        private static final long              serialVersionUID = 1L;

        protected ActorRef                      index            = null;
        protected ActorRef                      status           = null;
        protected models.spider.config.Factory  config           = null;
        protected models.spider.request.Factory fact             = null;

        SpiderFactory(final ActorRef index, final ActorRef status, final models.spider.config.Factory config, final models.spider.request.Factory fact) {
            this.config = config;
            this.index = index;
            this.status = status;
            this.fact = fact;
        }

        @Override
        public Actor create() {
            return new Spider(index, status, config, fact);
        }

    }

    /**
     * Creates index actors.
     * 
     * @author laurent@opprecht.info
     * 
     */
    public static class IndexFactory implements UntypedActorFactory {

        private static final long              serialVersionUID = 1L;

        protected models.spider.config.Factory config           = null;

        IndexFactory(final models.spider.config.Factory config) {
            this.config = config;
        }

        @Override
        public Actor create() {
            return new Index(config);
        }

    }

    /**
     * Creates Status actors.
     * 
     * @author laurent@opprecht.info
     * 
     */
    public static class StatusFactory implements UntypedActorFactory {

        private static final long serialVersionUID = 1L;

        StatusFactory() {
        }

        @Override
        public Actor create() {
            return new Status();
        }

    }

    public static class Message {

    }

    public static class Info extends Message {

    }
    
    public static class CountTh extends Message {

    }

    public static class Add extends Message {

        protected String url = null;

        Add(final String url) {
            this.url = url;
        }

        public String getUrl() {
            return url;
        }
    }

    public static class Refresh extends Message {

        protected String url = null;

        Refresh(final String url) {
            this.url = url;
        }

        public String getUrl() {
            return url;
        }
    }

    public static class Check extends Message {

    }
    
    public static class Crawl extends Message {

    }

    public static class Start extends Message {

        protected int threadCount = 1;

        Start(final int threadCount) {
            this.threadCount = Math.max(1, threadCount);
        }

        public int getThreadCount() {
            return threadCount;
        }

    }

    public static class Stop extends Message {

    }

    protected models.spider.config.Factory config;
    protected ActorRef                     index;
    protected ActorRef                     status;
    protected ActorRef                     spiders;
    protected Cancellable                  schedule = null;
    protected LoggingAdapter               log      = null;

    public Crawler(final models.spider.config.Factory config) {
        log = Logging.getLogger(getContext().system(), this);
        log.debug("crawler new");
        this.config = config;

        Props indexProperties = new Props(new IndexFactory(config));
        index = this.getContext().actorOf(indexProperties, "index");

        Props statusProperties = new Props(new StatusFactory());
        status = this.getContext().actorOf(statusProperties, "status");
    }

    ActorRef createSpiders(final int count) {
        Props spidersProperties = new Props(new SpiderFactory(index, status, config, config.newRequestFactory()));
        spidersProperties = spidersProperties.withRouter(new akka.routing.BroadcastRouter(count));
        ActorRef result = this.getContext().actorOf(spidersProperties);
        return result;
    }

    ActorRef spiders() {
        if (spiders != null && !spiders.isTerminated()) {
            return spiders;
        }
        spiders = createSpiders(1);
        return spiders;
    }

    @Override
    public void onReceive(final Object message) throws Exception {
        if (message instanceof Add) {
            add((Add) message);
        } else if (message instanceof Refresh) {
            refresh((Refresh) message);
        } else if (message instanceof Start) {
            start((Start) message);
        } else if (message instanceof Stop) {
            stop((Stop) message);
        } else if (message instanceof Check) {
            check((Check) message);
        } else if (message instanceof Info) {
            info((Info) message);
        } else if (message instanceof CountTh) {
            countTh((CountTh) message);
        } else if (message instanceof Crawl) {
            crawl((Crawl) message);
        } else {
            unhandled(message);
        }
    }

    protected void add(final Add message) {
        // log.debug("crawler add");
        index.tell(new Index.Add(message.getUrl()), getSelf());
    }

    protected void refresh(final Refresh message) {
        // log.debug("crawler refresh");
        index.tell(new Index.Refresh(message.getUrl()), getSelf());
    }

    protected void crawl(final Crawl message) {
        // log.debug("crawler crawl");
        spiders().tell(new Spider.Crawl(), getSelf());
    }

    protected void start(final Start message) {
        // log.debug("crawler start");
    	// --------- stop
        if (schedule != null) {
            schedule.cancel();
            schedule = null;
        }
        if (spiders != null && !spiders.isTerminated()) {
            context().stop(spiders);
            spiders = null;
        }
        // --------- end stop
        File.cleanFolder();
        spiders = createSpiders(message.getThreadCount());

        FiniteDuration start = FiniteDuration.create(0, TimeUnit.MILLISECONDS);
        FiniteDuration every = FiniteDuration.create(5, TimeUnit.MINUTES);
        schedule = Akka.system().scheduler().schedule(start, every, getSelf(), new Check(), Akka.system().dispatcher());
    }

    protected void stop(final Stop message) {
        // log.debug("crawler stop");
        if (schedule != null) {
            schedule.cancel();
        }
        if (spiders != null && !spiders.isTerminated()) {
            context().stop(spiders);
        }
        spiders = null;
        schedule = null;
    }

    protected void info(final Info message) {    	
    	status.forward(new Status.Info(), this.context());
    }
    
    protected void countTh(final CountTh message) {
    	spiders.forward(CurrentRoutees.getInstance(), this.context());
    }

    protected void check(final Check message) {
        spiders().tell(new Spider.Crawl(), getSelf());
    }

    @Override
    public void preStart() {
        // log.debug("crawler pre start");
    }

    @Override
    public void postStop() {
        // log.debug("crawler post stop");
    }
}
